package com.lagou.client;


import com.lagou.service.JSONSerializer;
import com.lagou.service.RpcEncoder;
import com.lagou.service.RpcRequest;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.io.InputStream;
import java.util.*;
import java.util.stream.Collectors;

public class UserClientUtil {


    private static Map<String, UserClientHandler> userClientHandlerMap = new HashMap<>();

    public static void loadZkProvider() throws Exception {

        //加载配置文件
        InputStream resourceAsStream = UserClientUtil.class.getClassLoader().getResourceAsStream("application.properties");
        Properties properties = new Properties();
        properties.load(resourceAsStream);
        String zkServerConnectUrl = properties.getProperty("zdy.zkServerConnectUrl");
        String serverZkBasePath = properties.getProperty("zdy.serverZkBasePath");


        //连接zookeeper服务器
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);
        CuratorFramework client = CuratorFrameworkFactory
                .builder().connectString(zkServerConnectUrl)
                .sessionTimeoutMs(1000)
                .connectionTimeoutMs(50000)
                .retryPolicy(retryPolicy)
                .build();

        client.start();


        //获取server节点
        List<String> serverNodeList = client.getChildren().forPath(serverZkBasePath);
        for (String serverNodePath : serverNodeList) {
            byte[] bytes = client.getData().forPath(serverZkBasePath + "/" + serverNodePath);
            String[] serverNodeInfo = new String(bytes).split(":");

            buildClientHandler(serverNodeInfo[0], Integer.parseInt(serverNodeInfo[1]));

        }

        //添加server节点变更监听器,监听server服务器的的加入和下线
        PathChildrenCache cache = new PathChildrenCache(client, serverZkBasePath, true);
        cache.start();

        PathChildrenCacheListener serverNodeListener = (server, event) -> {
            if (PathChildrenCacheEvent.Type.CHILD_ADDED.equals(event.getType())) {
                //检测到新节点加入，把新的server加入到处理器集合中
                String[] serverNodeInfo = new String(event.getData().getData()).split(":");

                System.out.println("有新的server节点加入，ip=" + serverNodeInfo[0] + ",port=" + serverNodeInfo[1]);
                buildClientHandler(serverNodeInfo[0], Integer.parseInt(serverNodeInfo[1]));
            } else if (PathChildrenCacheEvent.Type.CHILD_REMOVED.equals(event.getType())) {
                //检测到节点删除，把server从处理器集合中移除
                String[] serverNodeInfo = new String(event.getData().getData()).split(":");
                System.out.println("server节点下线，ip=" + serverNodeInfo[0] + ",port=" + serverNodeInfo[1]);
                userClientHandlerMap.remove(serverNodeInfo[0] + serverNodeInfo[1]);
            }
        };

        cache.getListenable().addListener(serverNodeListener);
    }

    private static void buildClientHandler(String serverIp, int port) throws InterruptedException {
        ServerInfo serverInfo = new ServerInfo(serverIp, port);

        UserClientHandler userClientHandler = initClient(serverInfo);
        userClientHandlerMap.put(serverIp + port, userClientHandler);
    }


    //初始化netty客户端
    private static UserClientHandler initClient(ServerInfo serverInfo) throws InterruptedException {
        final UserClientHandler userClientHandler = new UserClientHandler();
        userClientHandler.setServerInfo(serverInfo);

        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new StringDecoder());
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(userClientHandler);
                    }
                });

        bootstrap.connect(serverInfo.getIp(), serverInfo.getPort()).sync();
        return userClientHandler;
    }


    public static UserClientHandler getAvailableUserClientHandler() {
        if (userClientHandlerMap.isEmpty()) {
            return null;
        }

        List<ServerInfo> serverList = new ArrayList<>();
        for (String serverIpPort : userClientHandlerMap.keySet()) {
            ServerInfo serverInfo = userClientHandlerMap.get(serverIpPort).getServerInfo();
            long currentTimeMillis = System.currentTimeMillis();
            //判断最后一次响应时间是否超过5秒，超过5秒则清零
            if (currentTimeMillis - serverInfo.getLastCallTime() >= 5000) {
                serverInfo.setSpendTime(0);
            }

            serverList.add(serverInfo);
        }


        //按响应时间排序，选取响应时间最短的server
        List<ServerInfo> sortedServerList = serverList
                .stream().sorted(Comparator.comparing(ServerInfo::getSpendTime)).collect(Collectors.toList());

        ServerInfo fastestServer = sortedServerList.get(0);


        return userClientHandlerMap.get(fastestServer.getIp() + fastestServer.getPort());
    }
}
